package com.os.task;

import com.os.data.DataStore;
import com.os.config.RsConfig;
import com.os.feign.DataService;
import com.os.model.DataInfo;
import com.os.model.DataSource;
import com.os.model.LocationMsg;
import com.os.result.Resp;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.net.URI;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.stream.Collectors;

/**
 * 对分片数据存储
 * @author 风亦未止
 * @date 2023/5/4 11:34
 */
@Slf4j
public class PutShared implements Callable<Resp> {
    private DataService dataService;

    private DataInfo dataInfo;

    /**
     * 已上传的分片索引和所在的节点
     */
    private List<LocationMsg> locationMsgs;

    /**
     * 上传的数据
     */
    private List<DataSource> dataSources ;

    private RsConfig rsConfig;

    public PutShared(DataService dataService, DataInfo dataInfo, List<LocationMsg> locationMsgs, List<DataSource> dataSources, RsConfig rsConfig) {
        this.dataService = dataService;
        this.dataInfo = dataInfo;
        this.locationMsgs = locationMsgs;
        this.dataSources = dataSources;
        this.rsConfig = rsConfig;
    }

    @SneakyThrows
    @Override
    public Resp call() {
        List<String> excludeAddress = locationMsgs.stream().map(LocationMsg::getAddress).collect(Collectors.toList());
        List<Integer> excludeIndex = locationMsgs.stream().map(LocationMsg::getIndex).collect(Collectors.toList());
        List<URI> uris = DataStore.chooseRandomDataServer(rsConfig.TOTAL_SHARDS - locationMsgs.size(), excludeAddress);
        int cnt = 0;
        for(int i = 0; i < dataSources.size(); i++){
            DataSource dataSource = dataSources.get(i);
            assert dataSource.getOriginalFilename() != null;
            if(excludeIndex.contains(Integer.parseInt(dataSource.getOriginalFilename()))){
                continue;
            }
            URI uri = uris.get(cnt);
            log.info(String.format("节点%s写入数据~", uri));
            Resp put = dataService.put(uri, dataInfo.getHash(), dataSources.get(i));
            if(put.getCode() == 200){
                cnt++;
            }
        }
        if(cnt + locationMsgs.size() != rsConfig.TOTAL_SHARDS){
            return Resp.failed("上传失败~");
        }
        return Resp.success(null,"上传成功~");
    }
}
